package com.new_api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class FilterDemo2 {
    private Configuration conf;
    private Connection conn;
    private Admin admin;

    @Before
    public void init() throws IOException {
        // 创建配置文件
        conf = HBaseConfiguration.create();
        // 加载zookeeper的配置
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        // 获取连接
        conn = ConnectionFactory.createConnection(conf);
        // 获取admin对象
        admin = conn.getAdmin();
    }

    /**
     * 过滤出学号是以15001001开头的文科学生
     */
    @Test
    public void filterExec2() throws Exception {
// 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        // 过滤出学号是以15001001开头的文科学生
        Scan scan = new Scan();

        // 创建过滤器集合对象
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        // 创建过滤器
        PrefixFilter prefixFilter = new PrefixFilter("15001001".getBytes());

        //文科 的学生
        SingleColumnValueFilter clazzFilter = new SingleColumnValueFilter("info".getBytes(),
                "clazz".getBytes(), CompareOperator.EQUAL, new SubstringComparator("文科"));

        // 过滤器添加到filterList中，根据FilterList构造方法传入的 Operator 来决定多个filter的计算逻辑
        filterList.addFilter(prefixFilter);
        filterList.addFilter(clazzFilter);

        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        printScanner(scanner);

        table.close();

    }


    /**
     * 设置其实和结束rowkey
     */
    @Test
    public void filterExec1() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        // 通过运用4种比较器过滤出姓 于 ，年纪大于 21岁，性别为 女，且是 理科 的学生。
        Scan scan = new Scan();
        List<Filter> filters = new ArrayList<Filter>();

        SingleColumnValueFilter nameFilter = new SingleColumnValueFilter("info".getBytes(),
                "name".getBytes(), CompareOperator.EQUAL, new BinaryPrefixComparator("于".getBytes()));

        SingleColumnValueFilter ageFilter = new SingleColumnValueFilter("info".getBytes(),
                "age".getBytes(), CompareOperator.GREATER, new BinaryComparator("21".getBytes()));

        // 性别为 女
        SingleColumnValueFilter genderFilter = new SingleColumnValueFilter("info".getBytes(),
                "gender".getBytes(), CompareOperator.EQUAL, new BinaryComparator("女".getBytes()));

        //理科 的学生
        SingleColumnValueFilter clazzFilter = new SingleColumnValueFilter("info".getBytes(),
                "clazz".getBytes(), CompareOperator.EQUAL, new SubstringComparator("三班"));

        filters.add(nameFilter);
        filters.add(ageFilter);
        filters.add(genderFilter);
        filters.add(clazzFilter);

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);

        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);

        printScanner(scanner);

        table.close();
    }

    /**
     * 分页查询
     */
    @Test
    public void pageFilter() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        byte[] postfix = new byte[]{0x00};
        byte[] lastRow = null;
        Scan scan = new Scan();

        // 创建分页filter
        PageFilter pageFilter = new PageFilter(10);
        scan.setFilter(pageFilter);


        while (true) {
            if (lastRow != null) {
                scan.withStartRow(Bytes.add(lastRow, postfix));
            } else {
                scan.withStartRow(postfix);
            }

            // true 表示是最后一页
            boolean flag = true;

            ResultScanner scanner = table.getScanner(scan);

            int count = 0;
            for (Result result : scanner) {
                lastRow = result.getRow();
                flag = false;

                String id = new String(result.getRow());
                String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
                String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
                String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
                String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
                String ct = Bytes.toString(result.getValue("info".getBytes(), "ct".getBytes()));
                System.out.println("id:" + id + ",name:" + name + ",age:" + age + ",gender:" + gender
                        + ",clazz:" + clazz + ",ct:" + ct);
                count++;
            }
            System.out.println("本次scan了 " + count + " 行数据！！！");


            // 分页查询不到数据，跳出循环
            if (flag) {
                break;
            }
        }


//        table.close();

    }

    /**
     * 单列列值
     */
    @Test
    public void singleColumnValueFilter() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test", "student");
        // 获取table对象
        Table table = conn.getTable(tableName);

        Scan scan = new Scan();

        // 其实和结束的rowkey, 不包含stoprow
        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100010".getBytes());

//        SingleColumnValueFilter filter = new SingleColumnValueFilter("info".getBytes(),
//                "name".getBytes(), CompareOperator.EQUAL, "宣谷芹".getBytes());

//        SingleColumnValueFilter filter = new SingleColumnValueExcludeFilter("info".getBytes(),
//                "name".getBytes(), CompareOperator.EQUAL, "宣谷芹".getBytes());

//        PrefixFilter filter = new PrefixFilter("150010001".getBytes());

        // 只返回指定的列
//        ColumnPrefixFilter filter = new ColumnPrefixFilter("nam".getBytes());

        // 指定时间戳扫描
        // TimestampsFilter filter = new TimestampsFilter(Arrays.asList(1701158635999L, 1701158636999L));

        // 只返回指定的列
        FirstKeyOnlyFilter filter = new FirstKeyOnlyFilter();

        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);

        printScanner(scanner);

        table.close();
    }


    // 打印扫描结果
    private void printScanner(ResultScanner scanner) {
        int count = 0;
        for (Result result : scanner) {
            String id = new String(result.getRow());
            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            String ct = Bytes.toString(result.getValue("info".getBytes(), "ct".getBytes()));
            System.out.println("id:" + id + ",name:" + name + ",age:" + age + ",gender:" + gender
                    + ",clazz:" + clazz + ",ct:" + ct);
            count++;
        }
        System.out.println("本次scan了 " + count + " 行数据！！！");
    }


    @After
    public void afterTest() throws IOException {
        admin.close();
        conn.close();
    }
}
